package com.steven.logserver.init;

import com.steven.logserver.config.ZookeeperConfig;
import com.steven.logserver.handler.MyServerHandler;
import com.steven.logserver.mq.DirectMsgProducer;
import com.steven.logserver.protobuf.LogServerSendReqProto;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.RandomAccessFile;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.channels.FileChannel;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.CompletableFuture;

/**
 * 容器启动完成的时候初始化一个netty日志服务端，并注册到ZK
 *
 * @author qinxh
 * @date 2016/10/31
 */
@Component
@Order(2)
public class NettyLogServerInit implements ApplicationRunner {

    private static final Logger logger = LoggerFactory.getLogger(NettyLogServerInit.class);

    @Value("${log.netty.server.node.Path}")
    private String nodePath;
    @Value("${log.netty.bind.port}")
    private Integer bindPort;
    @Value("${log.netty.server.client-number}")
    private String clientNumberPath;
    @Value("${log.netty.server.file-folder}")
    private String fileFolder;
    @Value("${log.netty.zk.server.connection.timeout}")
    private Integer connectionTimeout;

    @Autowired
    private DirectMsgProducer producer;

    @Override
    public void run(ApplicationArguments args) throws InterruptedException {

        Thread.sleep(3000);
        CompletableFuture.runAsync(() -> {
            CuratorFramework client = ZookeeperConfig.getClient();
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            // 有连接到达时就会创建一个通道
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws UnknownHostException {
                                socketChannel.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                                socketChannel.pipeline().addLast(
                                        new ProtobufDecoder(LogServerSendReqProto.LogServerSendReq.getDefaultInstance()));
                                socketChannel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                                socketChannel.pipeline().addLast(new ProtobufEncoder());
                                socketChannel.pipeline().addLast(new MyServerHandler(bindPort, clientNumberPath, producer));
                            }
                        });
                logger.info("服务端准备就绪!");
                ChannelFuture channelFuture = bootstrap.bind(bindPort).sync();
                // 判断服务端启动成功后，将服务信息注册到zk
                if (channelFuture.isSuccess()) {
                    // 日志文件当前大小（byte） 作为为节点数据。如果日志文件不存在，则设置为0
                    String filePath = fileFolder + "/log_" + LocalDate.now().format(DateTimeFormatter.ofPattern("yyyy_MM_dd")) + ".file";
                    File file = new File(filePath);
                    byte[] data = "0".getBytes();
                    if(file.exists()){
                        RandomAccessFile reader = new RandomAccessFile(filePath, "r");
                        FileChannel channel = reader.getChannel();
                        data = String.valueOf(channel.size()).getBytes();
                        channel.close();
                        reader.close();
                    }
                    // 设定 ip + netty 服务绑定端口作为服务节点名
                    String path = nodePath + InetAddress.getLocalHost().getHostAddress() + ":" + bindPort;
                    client.create().creatingParentsIfNeeded()
                            //设置为临时节点，随着客户端SESSION消失而消失
                            .withMode(CreateMode.EPHEMERAL)
                            .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                            .forPath(path, data);
                }
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                logger.error("netty服务端准备异常：{}", e.getMessage());
            } catch (Exception e) {
                logger.error("netty服务端信息注册到zk异常：{}", e.getMessage());
            } finally {
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        });
    }
}
